////import java.util.{Date, UUID}
//
////import com.mongodb._
////import org.bson.Document
//import org.mongodb.scala.MongoClient
//import org.slf4j.LoggerFactory
//
////import scala.collection.convert.ImplicitConversions._
////import scala.collection.convert.ImplicitConversionsToJava
////import scala.util.{Random, Try}
//
//object mongodb {
//  lazy val log = LoggerFactory.getLogger(this.getClass)
//
//  //  def mongoSafeKey(pred: String) = pred.replace('.', '\u2025').replace('$', '\ufe69')
//
//  def main(args: Array[String]): Unit = {
//    MongoClient()
//      .getDatabase("techot")
//      .getCollection("out")
//      .count()
//      .subscribe((c:Long) => {println(c)})
//  }
//
//  //  def g(args: Array[String]) {
//  //    val blockSize = 1000000
//  //
//  //    val now = new Date()
//  //
//  //    log.info("loading per-evt-subjects ...")
//  //    val subjectPrefix = Try(io.Source.fromFile(
//  //      "%s/zjh/subject-per.s4per".format(System.getProperty("user.home"))
//  //    ).getLines()).getOrElse(Iterator.empty).zip(Iterator.continually("person-"))
//  //      .++(
//  //        Try(io.Source.fromFile(
//  //          "%s/zjh/subject-evt.s4evt".format(System.getProperty("user.home"))
//  //        ).getLines()).getOrElse(Iterator.empty).zip(Iterator.continually("event-"))
//  //      )
//  //      .toMap.withDefaultValue("other-")
//  //
//  //    log.info(s"done loading per-evt-subjects, size: ${subjectPrefix.size}")
//  //
//  //    log.info(s"connecting mongodb ...")
//  //
//  //    //    val mongoClient = new MongoClient()
//  //    val mongoClient = new MongoClient(new ServerAddress("localhost", 11001), List(MongoCredential.createScramSha1Credential("dbuser", "db", "secret.".toCharArray)))
//  //
//  //    val db = mongoClient.getDatabase("db")
//  //    //    val db = mongoClient.getDatabase("test")
//  //
//  //    val coll = db.getCollection("GlobalEntityLib")
//  //
//  //    val files = args.sortBy(x => Try(new java.io.File(x).length()).getOrElse(0L)) // sort by file size,
//  //      .toList
//  //
//  //    log.info(s"file inserting order: ${files.mkString(" ")}")
//  //
//  //    log.info("all ready...")
//  //
//  //    files
//  //      .par
//  //      .foreach(file => {
//  //        try {
//  //          insertFile(file)
//  //        } catch {
//  //          case e: Exception =>
//  //            log.error(s"err inserting file: $file", e)
//  //        }
//  //      })
//  //
//  //    def insertFile(file: String): Unit = {
//  //      log.info(s"in inserting file: $file")
//  //      val blocks = detectBlocks(file, blockSize)
//  //
//  //      val sv ="""^(".*")@(\w+)\s?\.""".r
//  //      val sv2 ="""^(".*")(?:\^\^<[^>]+>)?\s?\.""".r
//  //      val ev ="""^(\<.*\>)""".r
//  //
//  //
//  //      //      val lnc = new AtomicInteger(0)
//  //      //      val slnc = new AtomicInteger(0)
//  //      //      val dc = new AtomicInteger(0)
//  //
//  //      blocks
//  //        .par
//  //        .foreach(b => try {
//  //          insertBlock(file, b._1, b._2)
//  //        }
//  //        catch {
//  //          case e: Exception =>
//  //            log.error(s"err ")
//  //        })
//  //
//  //      def insertBlock(file: String, sa: Int, se: Int) = {
//  //        log.info(s"insert block slicing($sa, $se) on file: $file")
//  //
//  //        val docLns = io.Source.fromFile(file).getLines()
//  //          //        .drop(1) //仅drop 1 是不够的，最后一行也以#开头，中间行也有以#开头的
//  //          .slice(sa, se) // 根据detectBlock逻辑，这里的slice不可与filter交换
//  //          .filter(ln => ln.nonEmpty && ln.startsWith("#").unary_!)
//  //          .toList
//  //          .par
//  //
//  //        log.info(s"number of lines to process: ${docLns.size}")
//  //
//  //        docLns
//  //          .map(ln => {
//  //
//  //            //            lnc.incrementAndGet()
//  //
//  //            try {
//  //              val a = ln.split(" ", 3)
//  //              val s = a(0)
//  //              val p = a(1)
//  //              val vl = sv.findFirstMatchIn(a(2)).map(m => {
//  //                (m.group(1), m.group(2))
//  //              }).orElse(sv2.findFirstMatchIn(a(2)).map(m => {
//  //                (m.group(1), null)
//  //              })).orElse(ev.findFirstMatchIn(a(2)).map(m => {
//  //                //                (m.group(1), null)
//  //                //                log.warn(s"find relational object line: $lnc, file: $file")
//  //                log.warn(s"find relational object line, in file: $file")
//  //                null
//  //              })).getOrElse({
//  //                //                log.error(s"unexpected format, line $lnc: $ln, file: $file")
//  //                log.error(s"unexpected format, line: $ln, file: $file")
//  //                null
//  //              })
//  //
//  //              //            log.debug(s"vl for ${a(2)} is ${vl.toString()}")
//  //
//  //              //              slnc.incrementAndGet()
//  //
//  //              if (vl == null) {
//  //                //                log.error(s"no pattern rdf: ${a(2)}")  // 重复记录
//  //                null
//  //              } else {
//  //                (s, p, vl._1, vl._2)
//  //              }
//  //            } catch {
//  //              case e: Exception =>
//  //                log.error(s"on ln: $ln", e)
//  //                null
//  //            }
//  //          })
//  //          .filterNot(_ == null)
//  //          .groupBy(_._1)
//  //          .map(et => {
//  //            try {
//  //              val entity = et._1
//  //              val d = new Document()
//  //              val sPref = subjectPrefix(entity)
//  //              //            if (etype.nonEmpty) {
//  //              // {WARN}警告:会更新id字段,则实体关系必须后插
//  //              d.append("id", s"$sPref${UUID.randomUUID().toString}")
//  //              //            }
//  //              //              .append("entity", entity)
//  //              //              .append("id", s"person-${UUID.randomUUID()}").append("gmt_create", now)
//  //
//  //              val pts = et._2 // => list (s,p.o,lang)
//  //
//  //              pts
//  //                .groupBy(_._2) // groupby predicate => (predicate, list (a,p,o,lang) )
//  //                .mapValues(ts => {
//  //                val pds = ts
//  //                  // object is uniq
//  //                  //                // uniq value(rdf object)
//  //                  //                .groupBy(_._4)
//  //                  //                .values
//  //                  //                .map(_.head)
//  //                  .map(spol => {
//  //                  val obj = spol._3
//  //                  // cut "
//  //                  val pd = new Document().append("value", obj.substring(1, obj.length - 1)).append("timestamps", now)
//  //                  if (spol._4 != null) {
//  //                    pd.append("lang_type", spol._4)
//  //                  }
//  //                  pd
//  //                })
//  //                //              Array(pds.toList: _*)
//  //                ImplicitConversionsToJava.`seq AsJavaList`(pds.toList)
//  //              })
//  //                //              .toList
//  //                .foreach(pds => {
//  //                val pred = pds._1
//  //                //              val storedPred=pred.substring(pred.lastIndexOf('/') + 1
//  //                val storedPred = mongoSafeKey(pred)
//  //                // predicate: last char: >
//  //                d.append(storedPred, pds._2)
//  //              })
//  //
//  //              (entity, d)
//  //            } catch {
//  //              case e: Exception =>
//  //                log.error(s"errr mapping to mongo doc, (entity,tuple): ${et.toString()}")
//  //                null
//  //            }
//  //          })
//  //          .filter(_ == null)
//  //          .foreach(ed => {
//  //            val entity = ed._1
//  //            val doc = ed._2
//  //            try {
//  //              //              Try(log.debug(s"inserting doc: ${entity}, with props json: ${doc.toJson}"))
//  //
//  //              //              coll.insertOne(doc)
//  //              val ud = coll.updateOne(com.mongodb.client.model.Filters.eq("entity", entity), new Document("$set", doc))
//  //              //              val ud = coll.findOneAndUpdate(com.mongodb.client.model.Filters.eq("entity", entity), new Document("$set", doc))
//  //
//  //              //              Try(log.debug(s"after insert json: ${ud.toJson}"))
//  //
//  //              //              dc.incrementAndGet()
//  //
//  //              //              Try(log.debug(s"done inserting doc: $entity"))
//  //
//  //              if (Random.nextInt(1000000) == 0) {
//  //                // 1/1000000概率打印一次计数
//  //                Try(log.info(s"document count from collection.count(): ${coll.count()}"))
//  //              }
//  //            } catch {
//  //              case e: Exception =>
//  //                log.error(s"insert into mongo: ${doc.toJson()}", e)
//  //            }
//  //          })
//  //
//  //        log.info(s"done insert block done slicing($sa, $se) on file: $file")
//  //      }
//  //
//  //      //      log.info(s"done inserting file: $file, docs: ${dc.get()}, parsed lines: ${lnc.get()}, succ lines: ${slnc.get()}")
//  //      log.info(s"done inserting file: $file")
//  //    }
//  //
//  //    log.info(s"document final count: ${coll.count()}")
//  //
//  //
//  //    log.info("done main")
//  //  }
//
//
//  // 为了不让$blockSize把相同的主语、谓词记录分隔，需?要在blocksize附近找到主语间的边界
//  //may be iterator get a better performance on large file
//  //  def detectBlocks(file: String, blockSize: Int): List[(Int, Int)] = {
//  //    log.info(s"partion file: $file ...")
//  //
//  //    var itr = io.Source.fromFile(file).getLines()
//  //    var blks = 1
//  //    val bs = scala.collection.mutable.ListBuffer[(Int, Int)]()
//  //    var pre = 0
//  //    var ub = blks * blockSize
//  //    var ln = ""
//  //    do {
//  //      itr = itr.drop(blks * blockSize - pre)
//  //      ub = blks * blockSize
//  //      // meet EOF
//  //      if (itr.hasNext.unary_!) {
//  //        bs.append((pre, ub))
//  //      } else {
//  //        ln = itr.next()
//  //        ub += 1
//  //        // starts with # or following a EOF
//  //        if (ln.startsWith("#") || itr.hasNext.unary_!) {
//  //          bs.append((pre, ub))
//  //        } else {
//  //          // 这里的next一定有值
//  //          var ln2 = itr.next()
//  //          //            ub += 1
//  //          while (ln.split(" ")(0) == ln2.split(" ")(0)) {
//  //            ln2 = itr.next()
//  //            ub += 1
//  //          }
//  //          bs.append((pre, ub))
//  //        }
//  //      }
//  //      blks += 1
//  //      pre = ub
//  //    } while (itr.hasNext)
//  //
//  //    log.info(s"done partion file: $file")
//  //
//  //    bs.toList
//  //  }
//
//  //  def blocksItr(file: String, blockSize: Int): Iterator[(Int, Int)] ={
//  //    detectBlocks(file,blockSize).iterator
//  //  }
//}